package chapter3.eg1;

import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Before;
import org.junit.Test;

public class SendMessageApp {

  KafkaProducer<String, String> producer = null;

  @Before
  public void before() {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    producer = new KafkaProducer<>(properties);
  }

  @Test
  public void testSendAsync() {
    ProducerRecord<String, String> record = new ProducerRecord<>("CustomerFactory",
        "Precision Products", "France");
    try {
      producer.send(record);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  @Test
  public void testSendSync() {
    ProducerRecord<String, String> record = new ProducerRecord<>("CustomerFactory",
        "Precision Products", "France");
    try {
      RecordMetadata metadata = producer.send(record).get();
      System.out.println("offset: " + metadata.offset() + ", partition: " + metadata.partition());
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  @Test
  public void testSendWithCallback() throws InterruptedException {
    ProducerRecord<String, String> record = new ProducerRecord<>("CustomerFactory",
        "Precision Products", "France");
    producer.send(record, new Callback() {
      @Override
      public void onCompletion(RecordMetadata metadata, Exception e) {
        if (e != null) {
          e.printStackTrace();
          return;
        }

        System.out.println("offset: " + metadata.offset() + ", partition: " + metadata.partition());
      }
    });

    TimeUnit.SECONDS.sleep(5);
  }

}
